// 文件路径: dbms/execution/execution/LimitExecutor.java (这是一个新文件)
package com.dbms.execution.execution;

import com.dbms.common.Schema;
import com.dbms.common.Tuple;
import com.dbms.execution.plan.PlanNode;

/**
 * LimitExecutor (限制/偏移执行器) 负责处理 SQL 查询中的 LIMIT 和 OFFSET 子句。
 * 它的工作流程是：
 * 1. 在 `open()` 阶段，首先从子执行器中丢弃 (skip) `offset` 指定数量的行。
 * 2. 在 `next()` 阶段，继续从子执行器中获取行，但最多只返回 `limit` 指定数量的行。
 */
public class LimitExecutor implements Executor {
    /**
     * 子执行器，提供待处理的数据源。
     */
    private final Executor child;
    /**
     * 要返回的最大元组数量。
     */
    private final int limit;
    /**
     * 要跳过的初始元组数量。
     */
    private final int offset;
    /**
     * 已经产出（返回）的元组数量。
     */
    private int rowsProduced;

    public LimitExecutor(Executor child, int limit, int offset) {
        this.child = child;
        this.limit = limit;
        this.offset = offset;
    }

    @Override
    public void open() {
        child.open();
        this.rowsProduced = 0;

        // 【处理 OFFSET】
        // 在 open 阶段，循环调用子执行器的 next() 来跳过 offset 指定的行数
        for (int i = 0; i < offset; i++) {
            if (child.next() == null) {
                // 如果在跳过 offset 的过程中，子执行器就已经没有数据了，
                // 那么后续的 next() 调用将直接返回 null。
                break;
            }
        }
    }

    @Override
    public Tuple next() {
        // 【处理 LIMIT】
        // 如果已经产生的行数达到了 limit 的上限，则停止并返回 null
        if (rowsProduced >= limit) {
            return null;
        }

        // 从子执行器获取下一个元组
        Tuple tuple = child.next();
        if (tuple != null) {
            // 如果成功获取到元组，增加计数器并返回该元组
            rowsProduced++;
            return tuple;
        }

        // 如果子执行器没有更多元组了，返回 null
        return null;
    }

    @Override
    public void close() {
        child.close();
    }

    @Override
    public Schema getSchema() {
        // Limit 操作不改变 Schema，直接返回子执行器的 Schema
        return child.getSchema();
    }

    @Override
    public PlanNode getPlanNode() {
        // 为 EXPLAIN ANALYZE 创建计划节点
        String details = "limit: " + limit + ", offset: " + offset;
        PlanNode node = new PlanNode("Limit", details);
        node.setRowsProduced(this.rowsProduced);
        // 注意：LimitExecutor 本身的执行时间非常短，几乎可以忽略不计。
        // 主要的耗时都在其子节点中，因此我们这里不单独为 Limit 节点计时。
        // 递归地获取子执行器的计划节点，并将其添加为当前节点的子节点
        node.addChild(child.getPlanNode());
        return node;
    }
}